﻿using Flurl.Http;
using OkFlurl;
using System.IO.Compression;
using System.Runtime.CompilerServices;
using System.Text;
using System.Text.Encodings.Web;
using System.Text.Json;
using System.Text.Json.Serialization;

namespace VictoriaLogs.Client
{
    //https://docs.victoriametrics.com/victorialogs/data-ingestion/#json-stream-api
    //https://docs.victoriametrics.com/victorialogs/querying/
    public class VictoriaLogsClient
    {
        #region static

        public static readonly byte[] N = [(byte)'\n'];

        private static void InitStream(FlurlRequest req, string[] streams = null, string msgField = "_msg", string timeField = "_time")
        {
            req.WithHeader("Content-Type", "application/stream+json");
            req.SetQueryParam("_msg_field", msgField);
            req.SetQueryParam("_time_field", timeField);
            if (streams != null && streams.Length > 0)
            {
                if (streams.Length == 1)
                {
                    req.SetQueryParam("_stream_fields", streams[0]);
                }
                else
                {
#if NET462
                    var streamFields = string.Join(",", streams);
#else
                    var streamFields = string.Join(',', streams);
#endif
                    req.SetQueryParam("_stream_fields", streamFields);
                }

            }
        }

        //毫秒序列化
        public static readonly JsonSerializerOptions JsonOptionsMs = new()
        {
            Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,
            NumberHandling = JsonNumberHandling.AllowReadingFromString,
            PropertyNameCaseInsensitive = true,
            PropertyNamingPolicy = null,
            DictionaryKeyPolicy = null
        };

        static VictoriaLogsClient()
        {
            JsonOptionsMs.Converters.Add(new DateTimeMsConverter());
            JsonOptionsMs.Converters.Add(new DateTimeOffsetMsConverter());
        }

        #endregion

        #region MyRegion

        public FlurlRequest CreateRequest(string path)
        {
            var req = Api.CreateRequest(path);

            if (AccountID != 0)
            {
                req.WithHeader("AccountID", AccountID);
            }

            if (ProjectID != 0)
            {
                req.WithHeader("ProjectID", ProjectID);
            }

            return req;
        }

        #endregion

        public int AccountID { get; set; }

        public int ProjectID { get; set; }

        public FlurlApi Api { get; }

        public VictoriaLogsClient(string url, string username = null, string password = null, int? timeout = null)
        {
            Api = new FlurlApi(url)
            {
                UserName = username,
                Password = password,
                Timeout = timeout,
                JsonOptionsDefault = JsonOptionsMs
            };
        }

        public async Task InsertAsync(Func<Stream, Task> func, string[] streams = null, string msgField = "_msg", string timeField = "_time", bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, CancellationToken token = default)
        {
            var req = CreateRequest("/insert/jsonline");
            InitStream(req, streams, msgField, timeField);

            if (gzip)
            {
                req.WithHeader("Content-Encoding", "gzip");
            }

            using var fcs = new FlurlStreamContent(async stream =>
            {
                if (!gzip)
                {
                    await func(stream);
                    return;
                }

                using var gz = new GZipStream(stream, compressLv, true);
                await func(gz);
            });

            (await Api.PostAsync(req, fcs, token: token))?.Dispose();
        }

        public Task InsertListAsync<T>(IEnumerable<T> list, string[] streams = null, string msgField = "_msg", string timeField = "_time", bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default)
        {
            return InsertAsync(async stream =>
            {
                var i = 0;
                var currentRead = 0;
                byte[] bytes;
                foreach (var item in list)
                {
                    i++;
                    currentRead++;
                    if (i > 1)
                    {
#if NET462
                        await stream.WriteAsync(N, 0, 1, token);
#else
                        await stream.WriteAsync(N, token);
#endif
                    }

                    if (item is string str)
                    {
                        bytes = Encoding.UTF8.GetBytes(str);
#if NET462
                        await stream.WriteAsync(bytes, 0, bytes.Length, token);
#else
                        await stream.WriteAsync(bytes, token);
#endif
                    }
                    else
                    {
                        await JsonSerializer.SerializeAsync(stream, item, JsonOptionsMs, token);
                    }

                    if (currentRead == sendBuffer)
                    {
                        currentRead = 0;
                        await stream.FlushAsync(token);
                    }
                }

            }, streams, msgField, timeField, gzip, compressLv, token);
        }

        public async Task<Stream> QueryStreamAsync(string query, bool useForm = false, bool gzip = false, CancellationToken token = default)
        {
            var req = CreateRequest("/select/logsql/query");
            if (gzip)
            {
                req.WithHeader("Accept-Encoding", "gzip");
            }

            if (!useForm)
            {
                req.SetQueryParam("query", query);
                return await Api.GetStreamAsync(req, null, token);
            }
            else
            {
                return await Api.FormGetStreamAsync(req, new { query }, token: token);
            }
        }

        public async IAsyncEnumerable<string> QueryAsync(string query, bool useForm = false, bool gzip = false, [EnumeratorCancellation] CancellationToken token = default)
        {
            using var stream = await QueryStreamAsync(query, useForm, gzip, token);
            using var reader = new StreamReader(stream, Encoding.UTF8);
            string? line;
#if NET462 || NET6_0
            while ((line = await reader.ReadLineAsync()) != null)
            {
                token.ThrowIfCancellationRequested();
                yield return line;
            }
#else
            while ((line = await reader.ReadLineAsync(token)) != null)
            {
                yield return line;
            }
#endif
        }

        public async Task<Stream> QueryTailStreamAsync(string query, bool useForm = false, bool gzip = false, CancellationToken token = default)
        {
            var req = CreateRequest("/select/logsql/tail");
            if (gzip)
            {
                req.WithHeader("Accept-Encoding", "gzip");
            }

            if (!useForm)
            {
                req.SetQueryParam("query", query);
                return await Api.GetStreamAsync(req, null, token);
            }
            else
            {
                return await Api.FormGetStreamAsync(req, new { query }, token: token);
            }
        }

        public async IAsyncEnumerable<string> QueryTailAsync(string query, bool useForm = false, bool gzip = false, [EnumeratorCancellation] CancellationToken token = default)
        {
            using var stream = await QueryTailStreamAsync(query, useForm, gzip, token);
            using var reader = new StreamReader(stream, Encoding.UTF8);
            string? line;
#if NET462 || NET6_0
            while ((line = await reader.ReadLineAsync()) != null)
            {
                token.ThrowIfCancellationRequested();
                yield return line;
            }
#else
            while ((line = await reader.ReadLineAsync(token)) != null)
            {
                yield return line;
            }
#endif

        }

        public async Task<T> QueryOtherAsync<T>(QueryType queryType, string query, bool useForm = false, bool gzip = false, CancellationToken token = default)
        {
            string path = queryType switch
            {
                QueryType.Hits => "/select/logsql/hits",
                QueryType.StatsQuery => "/select/logsql/stats_query",
                QueryType.StatsQueryRange => "/select/logsql/stats_query_range",
                QueryType.StreamIds => "/select/logsql/stream_ids",
                QueryType.Streams => "/select/logsql/streams",
                QueryType.StreamFieldNames => "/select/logsql/stream_field_names",
                QueryType.StreamFieldValues => "/select/logsql/stream_field_values",
                QueryType.FieldNames => "/select/logsql/field_names",
                QueryType.FieldValues => "/select/logsql/field_values",
                _ => throw new Exception($"unknown {queryType}"),
            };

            var req = CreateRequest(path);
            if (gzip)
            {
                req.WithHeader("Accept-Encoding", "gzip");
            }

            if (!useForm)
            {
                req.SetQueryParam("query", query);
                return await Api.GetJsonAsync<T>(req, null, token);
            }
            else
            {
                return await Api.FormGetJsonAsync<T>(req, new { query }, token: token);
            }
        }
    }
}
